/****************************************************************
 * Licensed to the Apache Software Foundation (ASF) under one   *
 * or more contributor license agreements.  See the NOTICE file *
 * distributed with this work for additional information        *
 * regarding copyright ownership.  The ASF licenses this file   *
 * to you under the Apache License, Version 2.0 (the            *
 * "License"); you may not use this file except in compliance   *
 * with the License.  You may obtain a copy of the License at   *
 *                                                              *
 *   http://www.apache.org/licenses/LICENSE-2.0                 *
 *                                                              *
 * Unless required by applicable law or agreed to in writing,   *
 * software distributed under the License is distributed on an  *
 * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY       *
 * KIND, either express or implied.  See the License for the    *
 * specific language governing permissions and limitations      *
 * under the License.                                           *
 ****************************************************************/
package org.apache.james.mailbox.cassandra;

import java.nio.charset.CharacterCodingException;
import java.util.List;
import java.util.UUID;
import org.apache.cassandra.thrift.Cassandra;
import org.apache.cassandra.thrift.CfDef;
import org.apache.cassandra.thrift.KsDef;
import org.apache.cassandra.thrift.InvalidRequestException;
import org.apache.cassandra.thrift.NotFoundException;
import org.apache.cassandra.thrift.SchemaDisagreementException;
import org.apache.cassandra.thrift.TBinaryProtocol;
import org.apache.cassandra.utils.ByteBufferUtil;
import org.apache.james.mailbox.MailboxException;
import org.apache.james.mailbox.MailboxSession;
import org.apache.james.mailbox.SubscriptionException;
import org.apache.james.mailbox.cassandra.mail.CMailboxMapper;
import org.apache.james.mailbox.cassandra.mail.CMessageMapper;
import org.apache.james.mailbox.cassandra.user.CSubscriptionMapper;
import org.apache.james.mailbox.store.MailboxSessionMapperFactory;
import org.apache.james.mailbox.store.mail.MailboxMapper;
import org.apache.james.mailbox.store.mail.MessageMapper;
import org.apache.james.mailbox.store.mail.ModSeqProvider;
import org.apache.james.mailbox.store.mail.UidProvider;
import org.apache.james.mailbox.store.user.SubscriptionMapper;
import org.apache.thrift.TException;
import org.apache.thrift.protocol.TProtocol;
import org.apache.thrift.transport.TFramedTransport;
import org.apache.thrift.transport.TSocket;
import org.apache.thrift.transport.TTransport;
import org.apache.thrift.transport.TTransportException;


/**
 * HBase implementation of {@link MailboxSessionMapperFactory}
 * 
 */
public class CMailboxSessionMapperFactory extends MailboxSessionMapperFactory<UUID> {

    
    private UidProvider<UUID> uidProvider;
    private ModSeqProvider<UUID> modSeqProvider;
    private boolean cfexist=false;
    List<CfDef> lsCf ;
    
    
    public final boolean cfexistence(String cfname) throws TTransportException, InvalidRequestException, TException, NotFoundException
    {
     
                        TTransport transport = new TFramedTransport(new TSocket("localhost", 9160));
                        TProtocol protocol = new TBinaryProtocol(transport);
                        transport.open();
                        Cassandra.Client client = new Cassandra.Client(protocol);
                        client.set_keyspace(CNames.keyspace);
                        
                        KsDef kd=new KsDef();
                        
                        kd=client.describe_keyspace(CNames.keyspace);
                        
                        lsCf =kd.cf_defs;
                        
                        for(CfDef lsCf1 : lsCf)
                        {
                              if (lsCf1.getName().equalsIgnoreCase(cfname)) 
                              {
                                    return false;
                                    
                              }
               
                        }
                        
                        return true;
        }
    
    
    public CMailboxSessionMapperFactory(UidProvider<UUID> uidProvider, ModSeqProvider<UUID> modSeqProvider) throws TTransportException, InvalidRequestException, TException, CharacterCodingException, NotFoundException, SchemaDisagreementException {
     
                        this.uidProvider = uidProvider;
                        this.modSeqProvider = modSeqProvider;
                
                        
                        TTransport transport = new TFramedTransport(new TSocket("localhost", 9160));
                        TProtocol protocol = new TBinaryProtocol(transport);
                        transport.open();
                        Cassandra.Client client = new Cassandra.Client(protocol);
                        client.set_keyspace(CNames.keyspace);
                        
                        CfDef cd=new CfDef();
                           
                        if(cfexistence(ByteBufferUtil.string(CNames.MAILBOX_CF)))
                        {
                       
                                                 
                            cd.setKeyspace(CNames.keyspace);
                            cd.setName(ByteBufferUtil.string(CNames.MAILBOX_CF));
                            client.system_add_column_family(cd);
                            
                        }
                     
                        if (cfexistence(ByteBufferUtil.string(CNames.MESSAGES_META_CF))) 
                        {
                            cd.setKeyspace(CNames.keyspace);
                            cd.setName(ByteBufferUtil.string(CNames.MESSAGES_META_CF));
                            client.system_add_column_family(cd);
                            
                        }
                    
                        if(cfexistence(ByteBufferUtil.string(CNames.MESSAGE_DATA_BODY)))
                        {
                            cd.setKeyspace(CNames.keyspace);
                            cd.setName(ByteBufferUtil.string(CNames.MESSAGE_DATA_BODY));
                            client.system_add_column_family(cd);
                            
                        }
                        
                        if(cfexistence(ByteBufferUtil.string(CNames.MESSAGE_DATA_HEADERS)))
                        {
                    
                            cd.setKeyspace(CNames.keyspace);
                            cd.setName(ByteBufferUtil.string(CNames.MESSAGE_DATA_HEADERS));
                            client.system_add_column_family(cd);
                    
                         }
                        if (cfexistence(ByteBufferUtil.string(CNames.SUBSCRIPTION_CF))) 
                        {
                            cd.setKeyspace(CNames.keyspace);
                            cd.setName(ByteBufferUtil.string(CNames.SUBSCRIPTION_CF));
                            client.system_add_column_family(cd);
                        }
                        
    }
      

    @Override
    protected MessageMapper<UUID> createMessageMapper(MailboxSession session) throws MailboxException {
        return new CMessageMapper(session, uidProvider, modSeqProvider);
    }
     
     

    @Override
    protected MailboxMapper<UUID> createMailboxMapper(MailboxSession session) throws MailboxException {
        return new CMailboxMapper();
        
    }

    @Override
    protected SubscriptionMapper createSubscriptionMapper(MailboxSession session) throws SubscriptionException {
        return new CSubscriptionMapper();
    }

    public ModSeqProvider<UUID> getModSeqProvider() {
        return modSeqProvider;
    }
    public UidProvider<UUID> getUidProvider() {
        return uidProvider;
    }

}
